package com.sunzm.flink.datastream.java.operator;

import org.apache.commons.lang3.StringUtils;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

/**
 * JavaOperatorDemo
 *
 * @author Administrator
 * @version 1.0
 * @date 2021-06-22 20:46
 */
public class JavaOperatorDemo {
    private static boolean isLocal = true;

    public static void main(String[] args) throws Exception {
        //获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        if (isLocal) {
            env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        }

        DataStreamSource<String> dataStream = env.socketTextStream("82.156.210.70", 9999);

        DataStream<Tuple2<String, Integer>> pairDataStream = dataStream.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

            @Override
            public void flatMap(String line, Collector<Tuple2<String, Integer>> collector) throws Exception {
                if (StringUtils.isNotBlank(line)) {
                    //如果接收到的数据不是空，就开始进行单词切分
                    String[] fields = StringUtils.split(line, ",");

                    for (String word : fields) {
                        //输出每个单词
                        collector.collect(Tuple2.of(word, 1));
                    }
                }
            }

        });

        //KeySelector的第一个泛型是输入的数据的类型
        //第二个泛型是 最后输出的 key 的类型
        //匿名内部类
       /* KeyedStream<Tuple2<String, Integer>, String> keyedStream = pairDataStream.keyBy(new KeySelector<Tuple2<String, Integer>, String>() {

            @Override
            public String getKey(Tuple2<String, Integer> tp) throws Exception {
                //tp.f0就类似scala中的tp._1，这里就是取的单词
                return tp.f0;
            }

        });*/

        //lambd表达式
        KeyedStream<Tuple2<String, Integer>, String> keyedStream = pairDataStream.keyBy(tp -> tp.f0 );

        //按照第二个元素（次数）求和，得到每个单词的个数
        //keyedStream是一个按照key分组的datastream，求和运算是按照每个kei进行的
        DataStream<Tuple2<String, Integer>> wordCount = keyedStream.sum(1);

        wordCount.print();

        //启动程序
        String className = Thread.currentThread().getStackTrace()[1].getClassName();
        String simpleName = className.substring(className.lastIndexOf(".") + 1);
        env.execute(simpleName);
    }
}
